package com.yanxu;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Random;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-07-13-2025
 * @Description: api 学习
 *  keyBy 进行分区处理
 *
 */
public class Api09_keyby {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //源 -算子
        DataStreamSource<String> streamSource = env.fromElements("In the heart of the city stood an ancient library," +
                " its walls lined with dusty tomes and forgotten knowledge." +
                " The air inside was heavy with the scent of parchment and ink," +
                " a fragrance that whispered tales of bygone eras and unspoken wisdom. Among the stacks," +
                " a young woman named Elara spent her days, her fingers tracing the spines of books as she searched for a clue," +
                " a hint of the mystery that had brought her there. She was on a quest, not just for information, but for understanding," +
                " seeking to unravel the threads of history that had woven together to create the world she knew." +
                " Each book held a piece of the puzzle, each page turned brought her closer to the truth." +
                " But the library was more than just a repository of facts; it was a sanctuary," +
                " a place where the past and present intersected, where the whispers of the ancients mingled with the thoughts of the modern mind." +
                " And as Elara delved deeper into its depths, she discovered that some stories were not meant to be read, but experienced," +
                " for they held the power to change not just the course of history, but the very fabric of reality itself.");

        streamSource.print("from elements >>>>");

        //使用 flatmap 进行转换
        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> flatMapOperator = streamSource.flatMap(new FlatMapFunction<String, Tuple3<String, Integer, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple3<String, Integer, Integer>> collector) throws Exception {
                for (int i = 0; i < s.length(); i++) {
                    String str = s.substring(i, i + 1);
                    collector.collect(new Tuple3<>(str, 1, new Random().nextInt(10000)));
                }
            }
        });

        //根据 Tuple3 中的 f0 进行分组
        KeyedStream<Tuple3<String, Integer, Integer>, String> KeyedStream =
                flatMapOperator.keyBy(tuple -> tuple.f0);


        // KeyedStream<Tuple3<String, Integer, Integer>, Boolean> tuple3BooleanKeyedStream
        //        = flatMapOperator.keyBy(data -> true);

        //KeyedStream.window().aggregate()

        //分组之后，根据f1 进行sum
        KeyedStream.sum("f1").print("sum >>>");
        //分组之后，根据 f2 进行 maxBy
        KeyedStream.maxBy("f2").print("maxBy >>>");

        //对环境进行执行。
        env.execute();
    }

    /*
    * 常见的聚合算子
    *   sum
    *   max
    *   maxBy
    *   min
    *   minBy
    * */
}